package com.xmutca.service.a.web;

import org.junit.Test;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/test")
public class TestController {

    @GetMapping
    public String get() {
        return "finchley-service-a";
    }

    public static void main(String[] args) {
        Flux.just(1, 2, 3, 4, 5, 6);
        Mono.just(1);

        System.out.println(Flux.just());

        Mono.error(new Exception("some error"));

        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::println);
        Mono.just(1).subscribe(System.out::println);


        Mono<Long> count = Flux.just("flux", "mono")
                .flatMap(s -> Flux.fromArray(s.split("\\s*"))   // 1
                        // .delayElements(Duration.ofMillis(100)) // 2
                        .doOnNext(System.out::println)).doOnComplete(() -> System.out.println("xxx"))
                .count();

        System.out.println(count.block());
    }

    private Flux<String> getZipDescFlux() {
        String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
        return Flux.fromArray(desc.split("\\s+"));  // 1
    }

    @Test
    public void testSimpleOperators() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        getZipDescFlux().zipWith(Flux.interval(Duration.ofMillis(200))).subscribe(t -> System.out.println(t), t -> {

        }, countDownLatch::countDown);
        countDownLatch.await(10, TimeUnit.SECONDS);
    }

    private String getStringSync() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Hello, Reactor!";
    }

    @Test
    public void testSyncToAsync() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Mono.fromCallable(() -> getStringSync())    // 1
                .subscribeOn(Schedulers.elastic())  // 2
                .subscribe(System.out::println, null, countDownLatch::countDown);
        countDownLatch.await(10, TimeUnit.SECONDS);
    }
}
